/*
 *  Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved
 *
 *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 *  Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
 *  use this file except in compliance with the License. A copy of the License is
 *  located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 *  or in the "license" file accompanying this file. This file is distributed on
 *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 *  express or implied. See the License for the specific language governing
 *  permissions and limitations under the License.
 */

package io.temporal.samples.encodefailures;

import com.google.protobuf.ByteString;
import io.temporal.api.common.v1.Payload;
import io.temporal.payload.codec.PayloadCodec;
import io.temporal.payload.codec.PayloadCodecException;
import java.util.List;
import java.util.stream.Collectors;
import org.jetbrains.annotations.NotNull;

/**
 * Simple codec that adds dummy prefix to payload. For this sample it's also applied for failure
 * messages.
 */
public class SimplePrefixPayloadCodec implements PayloadCodec {

  public static final ByteString PREFIX = ByteString.copyFromUtf8("Customer: ");

  @NotNull
  @Override
  public List<Payload> encode(@NotNull List<Payload> payloads) {
    return payloads.stream().map(this::encode).collect(Collectors.toList());
  }

  private Payload encode(Payload decodedPayload) {
    ByteString encodedData = PREFIX.concat(decodedPayload.getData());
    return decodedPayload.toBuilder().setData(encodedData).build();
  }

  @NotNull
  @Override
  public List<Payload> decode(@NotNull List<Payload> payloads) {
    return payloads.stream().map(this::decode).collect(Collectors.toList());
  }

  private Payload decode(Payload encodedPayload) {
    ByteString encodedData = encodedPayload.getData();
    if (!encodedData.startsWith(PREFIX))
      throw new PayloadCodecException("Payload is not correctly encoded");
    ByteString decodedData = encodedData.substring(PREFIX.size());
    return encodedPayload.toBuilder().setData(decodedData).build();
  }
}
